package cn.cas.mango.util.singlepass;

import cn.cas.mango.dto.HotSpotNews;
import cn.cas.mango.util.Matrix;
import cn.cas.mango.util.ThreadUtil.ThreadPoolUtils;
import org.elasticsearch.common.collect.Tuple;


import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class SinglePass {
  private List<List<HotSpotNews>> lists;
  private double aggWeight;
  private Integer vectorSize;
  private String vectorField;
  private Map<Integer, List<List<HotSpotNews>>> map;

  public SinglePass(List<List<HotSpotNews>> lists, double aggWeight, Integer vectorSize, String vectorField) {
    this.lists = lists;
    this.aggWeight = aggWeight;
    this.vectorSize = vectorSize;
    this.vectorField = vectorField;
    this.map = new ConcurrentHashMap<>();
  }

  public  List<List<HotSpotNews>> getClusters() {
    List<List<HotSpotNews>> ret = new ArrayList<>();
    int cpuNum = Runtime.getRuntime().availableProcessors();
    int coreSize = cpuNum * 2;

    List<Tuple<Integer, Integer>> batches = makeBatch(0, lists.size() - 1, coreSize - 1);
    CountDownLatch startGate = new CountDownLatch(batches.size());
    AtomicInteger count = new AtomicInteger();
    for (Tuple<Integer, Integer> batch: batches) {
      List<List<HotSpotNews>> temp = new ArrayList<>();
      for (int i = batch.v1(); i < batch.v2(); i++) {
        temp.add(lists.get(i));
      }
      ThreadPoolUtils.execute(()->batchSinglePass(temp, aggWeight, startGate, count.getAndIncrement()));
    }
    try {
      startGate.await();
    } catch (InterruptedException ex) {
      throw new RuntimeException(ex);
    }
    for (List<List<HotSpotNews>> list: map.values()) {
      ret.addAll(list);
    }
//    for (List<HotSpotNews> list: lists) {
//      List<List<HotSpotNews>> cluster = singlePass(list, aggWeight);
//      ret.addAll(cluster);
//    }
    ret.sort((x,y)->-Integer.valueOf(x.size()).compareTo(y.size()));
    return ret;
  }

  private List<Tuple<Integer, Integer>> makeBatch(int start, int end, int splitNum) {
    List<Tuple<Integer, Integer>> ret = new ArrayList<>();
    int range = (end - start + 1) / splitNum;
    for (int i = 0 ; i < splitNum; i++) {
      ret.add(Tuple.tuple(i * range, i * range + range));
    }
    int left = end - start + 1 - range * splitNum;
    if (left != 0) {
      ret.add(Tuple.tuple(end + 1 - left, end + 1));
    }
    return ret;
  }
  public void batchSinglePass(List<List<HotSpotNews>> newsList, double threshold, CountDownLatch startGate, int order) {
    List<List<HotSpotNews>> ret  = new ArrayList<>();
    for (List<HotSpotNews> list: newsList) {
      List<List<HotSpotNews>> cluster = singlePass(list, threshold);
      ret.addAll(cluster);
    }
    map.put(order, ret);
    startGate.countDown();
  }


  public List<List<HotSpotNews>> singlePass(List<HotSpotNews> newsList, double threshold) {
    List<List<HotSpotNews>> ret = new ArrayList<>();
    for (int i = 0; i < newsList.size(); i++) {
      boolean flag = false;
      HotSpotNews now = newsList.get(i);
      for (int j = 0; j < ret.size(); j++) {
        List<HotSpotNews> cluster = ret.get(j);
        List<Double> avg;
        List<Double> vector;
        if (vectorField.equals("t")) {
           avg = Matrix.getAvg(cluster.stream()
                  .map(x -> (List<Double>) x.getVector()).collect(Collectors.toList()),
              vectorSize);
          vector = (List<Double>)now.getVector();
        } else if (vectorField.equals("c")) {
          avg = Matrix.getAvg(cluster.stream()
                  .map(x -> (List<Double>) x.getContentVector()).collect(Collectors.toList()),
              vectorSize);
          vector = (List<Double>)now.getContentVector();
        } else {
          avg = Matrix.getAvg(cluster.stream()
                  .map(x -> (List<Double>) x.getTitleContentVector()).collect(Collectors.toList()),
              vectorSize);
          vector = (List<Double>)now.getTitleContentVector();
        }
        double v = Matrix.cosineSimilarity(avg, vector);
        if (v > threshold) {
          cluster.add(now);
          flag = true;
          break;
        }
      }
      if (!flag) {
        List<HotSpotNews> temp = new ArrayList<>();
        temp.add(now);
        ret.add(temp);
      }

    }
    return ret;
  }

}
